
#ifndef __BACKUPBASE_H__
#define __BACKUPBASE_H__

#include "vcsbase.h"
#include "mbinterface.h"
#include "RtInetAddr.h"
#include "mrpdu.h"

class BackupModuleSink
{
public:
	virtual ~BackupModuleSink() {}
	virtual void OnActivate(BOOL bFailover) = 0;
	virtual void OnDeactivate() = 0;
};

template<class T>
class CBackupBase
{
public:
	virtual RtResult Initialize(
		const CRtString& strPeerIP, 
		WORD nPeerPort, 
		const CRtString& strGroupName,
		const CRtString& strClusterID ,
		BackupModuleSink* pSink)
	{
		RtResult rv = RT_OK;
		m_pSink = pSink;
		m_dwPeerNodeID = 0;

		m_pMsgBus = IMessageBus::Instance();
		if (NULL==m_pMsgBus) 
			return RT_ERROR_NULL_POINTER;

		m_dwNodeID = m_pMsgBus->GetLocalNodeID();
		if (m_dwNodeID==0)
			return RT_ERROR_FAILURE;

		m_strPeerIP.assign(strPeerIP);
		m_nPeerPort = nPeerPort;
		if (!m_strPeerIP.empty()) {
			CRtInetAddr peerAddr(m_strPeerIP.c_str(), m_nPeerPort);
			rv = m_pMsgBus->AddPeer(peerAddr);
			if (RT_FAILED(rv)) {
				RT_ERROR_TRACE("CGcBackupBase::Initialize, message bus AddPeer failed! group="<<m_strGroupName<<" rv="<<rv);
				return rv;
			}

			rv = m_pMsgBus->JoinGroup(strGroupName, GROUP_TYPE_SN_ONLY);
			if (RT_FAILED(rv)) {
				RT_ERROR_TRACE("CGcBackupBase::Initialize, message bus JoinGroup failed! group="<<m_strGroupName<<" rv="<<rv);
				return rv;
			}
		}

		m_strGroupName.assign(strGroupName);
		m_strClusterID.assign(strClusterID);
		m_dwPeerDownTick = GetTickCountEx();
		return RT_OK;
	}

	virtual  void Release(){
		if (m_pMsgBus) {
			if (!m_strPeerIP.empty()) {
				m_pMsgBus->LeaveGroup(m_strGroupName);
				CRtInetAddr peerAddr(m_strPeerIP.c_str(), m_nPeerPort);
				m_pMsgBus->RemovePeer(peerAddr);
			}
			m_pMsgBus = NULL;
		}
		m_pSink = NULL;
		m_dwNodeID = 0;
		m_dwPeerNodeID = 0;
	}

	virtual  void OnNodeJoin(DWORD dwNodeID){}

	virtual  void OnNodeLeave(DWORD dwNodeID){
		if (dwNodeID==m_dwPeerNodeID) {
			m_dwPeerNodeID = 0;
			m_dwPeerDownTick = GetTickCountEx();
		}
	}

	void OnMessage(DWORD dwNodeID, CRtMessageBlock& msg){
		RT_ASSERTE(m_dwPeerNodeID==dwNodeID);
		CMrPduBase* pdu = NULL;
		CMrPduDecoder::DecodePdu(msg, pdu);
		if (NULL==pdu)
			return;

		WORD nType = pdu->GetType();
		static_cast<T*>(this)->HandlePdu(dwNodeID, nType, *pdu);
		delete pdu;
	}

	DWORD GetNodeID() const { return m_dwNodeID; }
	DWORD GetPeerNodeID() const { return m_dwPeerNodeID; }
	const CRtString& GetPeerIP() const { return m_strPeerIP; }
	const CRtString& GetClusterID() const { return m_strClusterID; }
	void OnControllerTimer(){
		LONGLONG dwCurrentTick = GetTickCountEx();
		static_cast<T*>(this)->OnControllerTimer_i(dwCurrentTick);
	}

	RtResult SendPdu(const CMrPduBase& pdu){
		CRtMessageBlock mb(pdu.GetLength());
		RtResult nResult = pdu.Encode(mb);
		if (RT_FAILED(nResult)) {
			RT_WARNING_TRACE("CBackupBase::SendPdu, pdu StreamTo failed! group="<<m_strGroupName<<" PduType="<<pdu.GetType()<<" nResult="<<nResult);
			return nResult;
		}
		if (m_pMsgBus && m_dwPeerNodeID>0)
			m_pMsgBus->Unicast(m_dwPeerNodeID, mb, TRUE);
		return RT_OK;
	}

protected:
	CBackupBase()
	: m_pSink(NULL)
	, m_pMsgBus(NULL)
	, m_dwNodeID(0)
	, m_dwPeerNodeID(0)
	, m_dwPeerDownTick(0)
	{
	}

	virtual ~CBackupBase(){
		Release();
	}

protected:
	BackupModuleSink*	m_pSink;
	IMessageBus*		m_pMsgBus;
	CRtString			m_strPeerIP;
	WORD				m_nPeerPort;
	CRtString			m_strGroupName;
	CRtString			m_strClusterID;
	DWORD				m_dwNodeID;
	DWORD				m_dwPeerNodeID;
	LONGLONG			m_dwPeerDownTick;
};

#endif // __RT_BACKUPBASE_H__
